/*
 * MemPageBoundedBufferVoidKeyImpl_.h
 *
 *  Created on: Aug 1, 2011
 *      Author: yong
 */

#ifndef MEMPAGEBOUNDEDBUFFERVOIDKEYIMPL__H_
#define MEMPAGEBOUNDEDBUFFERVOIDKEYIMPL__H_
#include<vector>
#include<cstdlib>
#include<boost/utility.hpp>
#include"../Typedef.h"
#include "../Config.h"

namespace distrim
{
/**
 * @brief  No key, fixed value size.
 */
template<typename V, typename P>
class MemPageBoundedBuffer<void, V, P, true, true> : private boost::noncopyable
{

private:
	struct PageInfo // Inner class to save state of buffers.
	{
		PageInfo() :
			PageBegin(NULL), State(StorageState::Empty), IsTaken(false)
		{
		}
		void * PageBegin;
		volatile StorageState::StorageStateEnum State;
		volatile bool IsTaken;
	};
public:
	typedef volatile size_t index_type;

public:
	/**
	 * @param initBufferCount  initial buffer number
	 * @throws  std::runtime_error if memory allocation failed.
	 *
	 * @brief  Construct a bounded buffer with given number of maximum values,
	 * without key.
	 */
	MemPageBoundedBuffer(Config config, CacheMode::CacheModeEnum cacheMode)
			throw (std::runtime_error) :
		m_config(config), m_vecBuffer(m_config.GetInputBufferCount()),
				m_allocableBuffer(m_config.GetInputBufferCount()),
				m_filledBuffer(0), m_producedBuffer(0), m_consumedBuffer(0),
				m_pageBytes(sizeof(V) * m_config.GetValueElemCount()
						* m_config.GetPageCapacity()), m_cacheMode(cacheMode)
	{
		for (size_t i = 0, s = m_vecBuffer.size(); i < s; ++i)
		{
			m_vecBuffer[i].PageBegin = malloc(m_pageBytes);
			if (m_vecBuffer[i].PageBegin == NULL)
			{
				for (size_t j = 0; j < i; ++j)
				{
					free(m_vecBuffer[i].PageBegin);
				}
				BOOST_THROW_EXCEPTION(std::runtime_error("error allocating space"));
			}
		}
	}
private:
	Config m_config;
	std::vector<PageInfo> m_vecBuffer;
	volatile size_t m_allocableBuffer, m_filledBuffer, m_producedBuffer,
			m_consumedBuffer;
	Mutex m_filledBufferMutex, m_allocatableBufferMutex;
	Condition m_bufferNoneToAllocateCondition, m_bufferNoneFilledCondition;
	const size_t m_pageBytes;
	CacheMode::CacheModeEnum m_cacheMode;

public:
	inline void Reset()
	{
		m_producedBuffer = 0;
		m_consumedBuffer = 0;
	}
	/**
	 * Recycle buffers allocated.
	 */
	virtual ~MemPageBoundedBuffer()
	{
		for (size_t i = 0, s = m_vecBuffer.size(); i < s; ++i)
		{
			free(m_vecBuffer[i].PageBegin);
		}
	}

	/**
	 * @return  return a valid index of buffer to be filled in.
	 * @brief  Get a valid buffer to fill data in.
	 */
	index_type MutexGetIndexToFill()
	{
		if (m_cacheMode != CacheMode::BoundedCache)
		{
			BOOST_THROW_EXCEPTION(std::logic_error("error working mode"));
		}
		size_t fileSplits = m_config.GetFileSplitsCount();
		// Wait until find.
		while (true)
		{
			Lock lk(m_allocatableBufferMutex);
			if (m_producedBuffer < fileSplits)
			{
				if (m_allocableBuffer > 0)
				{
					for (size_t i = 0, s = m_vecBuffer.size(); i < s; ++i)
					{
						if (m_vecBuffer[i].State == StorageState::Empty)
						{
							m_vecBuffer[i].State = StorageState::Allocated;
							--m_allocableBuffer;
							++m_producedBuffer;
							return i;
						}
					}
				}
				else
				{
					m_bufferNoneToAllocateCondition.wait(lk);
				}
			}
			else
			{
				m_bufferNoneToAllocateCondition.notify_all();// Notify all other sleeping threads
				// slept due to allocatable buffer == 0.
				return -1;
			}
		}
	}

	/**
	 * @param  indexToFill  Index of buffer to fill in.
	 * @param  kvPage  The page that associates with the file to be read in.
	 */
	void MutexFillBuffer(index_type indexToFill,
			KeyValuePage<void, V, P> &kvPage)
	{
		if (m_cacheMode != CacheMode::BoundedCache)
		{
			BOOST_THROW_EXCEPTION(std::logic_error("error working mode"));
		}
		// Fill buffer allocated.
		kvPage.Page<void, V>::SetPageBegin(m_vecBuffer[indexToFill].PageBegin);
		kvPage.Produce();
		Lock lk(m_filledBufferMutex);
		{
			m_vecBuffer[indexToFill].State = StorageState::Filled;
			if (++m_filledBuffer == 1)
			{
				m_bufferNoneFilledCondition.notify_one(); // Notify a sleeping consumer.
			}
		}
	}

	/**
	 * @return  Get index of an empty buffer.
	 * @note  If all buffer have been processed, return -1.
	 */
	index_type MutexGetIndexToEmpty()
	{
		if (m_cacheMode != CacheMode::BoundedCache)
		{
			BOOST_THROW_EXCEPTION(std::logic_error("error working mode"));
		}
		while (true)
		{
			Lock lk(m_filledBufferMutex);
			if (m_consumedBuffer < m_config.GetFileSplitsCount())
			{
				if (m_filledBuffer > 0)
				{
					for (size_t i = 0, s = m_vecBuffer.size(); i < s; ++i)
					{
						if (m_vecBuffer[i].State == StorageState::Filled
								&& m_vecBuffer[i].IsTaken == false)
						{
							m_vecBuffer[i].IsTaken = true;
							--m_filledBuffer;
							++m_consumedBuffer;
							return i;
						}
					}
				}
				else
				{
					m_bufferNoneFilledCondition.wait(lk);
				}
			}
			else // Return -1 indicates that every buffer has been processed.
			{
				m_bufferNoneFilledCondition.notify_all(); // Notify all other sleeping threads
				// slept due to filled buffer == 0.
				return -1;
			}
		}
	}

	/**
	 * @param  indexToEmpty  Index of buffer to consume.
	 * @param  kvPage  The page that associates with how the buffer should be consumed.
	 */
	void MutexConsumeBuffer(index_type indexToEmpty,
			KeyValuePage<void, V, P> &kvPage)
	{
		if (m_cacheMode != CacheMode::BoundedCache)
		{
			BOOST_THROW_EXCEPTION(std::logic_error("error working mode"));
		}
		kvPage.Page<void, V>::SetPageBegin(m_vecBuffer[indexToEmpty].PageBegin);
		kvPage.Consume();
		{
			Lock lk(m_filledBufferMutex);
			m_vecBuffer[indexToEmpty].State = StorageState::Empty;
			m_vecBuffer[indexToEmpty].IsTaken = false;
		}
		Lock lk(m_allocatableBufferMutex);
		if (++m_allocableBuffer == 1)
		{
			m_bufferNoneToAllocateCondition.notify_one();
		}
	}
	/**
	 * @param
	 */
	void DirectFillBuffer(index_type indexToFill,
			KeyValuePage<void, V, P> &kvPage)
	{
		kvPage.Page<void, V>::SetPageBegin(m_vecBuffer[indexToFill].PageBegin,
				false);
		kvPage.Produce();
		m_vecBuffer[indexToFill].State = StorageState::Filled;
	}
	/**
	 *
	 */
	size_t Size() const
	{
		return m_vecBuffer.size();
	}
	size_t size() const
	{
		return m_vecBuffer.size();
	}
	const PageInfo &operator[](index_type index) const
	{
		return m_vecBuffer[index];
	}

	PageInfo &operator[](index_type index)
	{
		return m_vecBuffer[index];
	}

	const PageInfo &At(index_type index) const
	{
		if (index >= m_vecBuffer.size())
		{
			BOOST_THROW_EXCEPTION(std::invalid_argument("index out of range"));
		}
		return operator[](index);
	}

	PageInfo &At(index_type index)
	{
		if (index >= m_vecBuffer.size())
		{
			BOOST_THROW_EXCEPTION(std::invalid_argument("index out of range"));
		}
		return operator[](index);
	}

};
}

#endif /* MEMPAGEBOUNDEDBUFFERVOIDKEYIMPL__H_ */
